package com.leyantech.ecp.agenx.session.client;

import com.leyantech.utility.concurrent.DynamicThreadPools;
import com.leyantech.utility.constructs.Constructs;

import com.ctrip.framework.apollo.Config;
import com.ctrip.framework.apollo.ConfigService;
import com.google.common.cache.CacheLoader;
import com.google.common.collect.Maps;
import com.google.inject.AbstractModule;
import com.google.inject.Inject;
import com.google.inject.Provider;
import com.google.inject.TypeLiteral;
import com.google.inject.name.Named;
import com.google.inject.name.Names;
import io.opentracing.contrib.kafka.TracingKafkaConsumer;
import io.opentracing.util.GlobalTracer;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Collections;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-11-17.
 */
public class SessionClientModule extends AbstractModule {

  @Override
  protected void configure() {

    final Named named = Names.named("session-client");

    bind(new TypeLiteral<CacheLoader<String, Set<Assistant>>>() {
    })
        .annotatedWith(named)
        .toProvider(getCacheLoaderProvider())
        .asEagerSingleton();

    final Consumer<String, String> consumer = provideConsumer();

    bind(new TypeLiteral<Consumer<String, String>>() {
    })
        .annotatedWith(named)
        .toInstance(consumer);

    final ExecutorService executorService = executorService();

    bind(ExecutorService.class).annotatedWith(named).toInstance(executorService);

    Runtime.getRuntime()
        .addShutdownHook(new Thread(() -> getProvider(SessionClient.class).get().stop()));

  }

  protected Provider<CacheLoader<String, Set<Assistant>>> getCacheLoaderProvider() {

    return new Provider<CacheLoader<String, Set<Assistant>>>() {

      @Inject
      SessionManagerLoader sessionManagerLoader;

      @Override
      public CacheLoader<String, Set<Assistant>> get() {
        return CacheLoader.from(sessionManagerLoader::load);
      }
    };

  }

  protected Config apolloConfig() {
    return ConfigService.getConfig("1.ecp-agenx-pdd");
  }

  protected Map<String, Object> consumerConfig() {
    final Map<String, Object> config = Maps.newHashMap();

    final String clientId = String.join(".",
        Constructs.LAIN_APPNAME,
        Constructs.HOSTNAME
    );

    final String brokerServer = apolloConfig().getProperty("broker.list", "");

    config.put("max.poll.records", 100);
    config.put("bootstrap.servers", brokerServer);
    config.put("enable.auto.commit", "true");
    config.put("client.id", clientId);
    config.put("auto.offset.reset", "latest");
    // 每台实例独自消费
    config.put("group.id", clientId);

    config.put("key.deserializer", StringDeserializer.class);
    config.put("value.deserializer", StringDeserializer.class);

    return config;
  }

  protected Consumer<String, String> provideConsumer() {

    final Consumer<String, String> consumer = new KafkaConsumer<>(consumerConfig());

    final String topic = apolloConfig().getProperty("assistant-event.topic", "");
    consumer.subscribe(Collections.singletonList(topic));

    return new TracingKafkaConsumer<>(consumer, GlobalTracer.get());
  }

  protected ExecutorService executorService() {
    return DynamicThreadPools.create("assistant-event-consumer", 1, 10, 60, "sync", 0, true);
  }

}
